package com.inns.position.handler;

import com.inns.position.hbase.service.IDataService;
import com.inns.position.kafka.bean.LocationInfo;
import com.inns.position.redis.service.RealTimeDataService;
import com.inns.position.utils.DateUtil;
import com.inns.position.utils.TimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by phoenix on 2018/4/21.
 */
@Component
public class DefaultHandler {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private RealTimeDataService realTimeDataService;

    @Resource
    private IDataService iDataService;

    private static final int MAX_STOP_TIME = 30000;
    private volatile Boolean isRunning = true;
    private int blockingQueueSize = 50000;
    private static final int nThreads = Runtime.getRuntime().availableProcessors() * 4;

    private BlockingQueue<LocationInfo> blockingQueue = new ArrayBlockingQueue<>(blockingQueueSize);

    private final AtomicInteger threadId = new AtomicInteger(0);
    private final AtomicLong lastLogTime = new AtomicLong();// 下次打印日志时间（为了调试方便，每隔固定时间打印日志）

    private ThreadPoolExecutor executorService = (ThreadPoolExecutor) (Executors.newFixedThreadPool(nThreads, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("default_handler_" + threadId.incrementAndGet());
            return thread;
        }
    }));

    public void addToQueue(LocationInfo locationInfo) {
        try {
            boolean offer = blockingQueue.offer(locationInfo, 1, TimeUnit.SECONDS);
            if (!offer) {
                logger.error("blockingQueue.offer failed,locationInfo:" + locationInfo);
            }
        } catch (InterruptedException e) {
            logger.error("occured InterruptedException", e);
        }
    }

    @PostConstruct
    public void notifyApp() {
        shutdownHook();

        int availableProcessors = Runtime.getRuntime().availableProcessors();
        for (int i = 0; i < availableProcessors; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    while (isRunning) {
                        LocationInfo locationInfo = null;
                        try {
                            locationInfo = blockingQueue.poll(1000, TimeUnit.MILLISECONDS);
                            if (locationInfo == null) {
                                continue;
                            }
                            saveInfo(locationInfo);
                        } catch (InterruptedException e) {
                            logger.error("notify failed.", e);
                            return;
                        } catch (Exception e) {
                            logger.error("notify locationInfo:" + locationInfo + " failed.", e);
                        }
                    }
                }
            });
        }
    }

    private void saveInfo(LocationInfo locationInfo) {
        long curTime = System.currentTimeMillis();
        boolean logTag = false;
        if (lastLogTime.get() < curTime) {
            lastLogTime.set(curTime + 10 * 1000);
            logTag = true;
            log("blockingQueue size=" + blockingQueue.size() + ", " + locationInfo.toString());
        }
        try {
            iDataService.save(locationInfo);
        } catch (Exception e) {
            logger.error("hbase saveInfo error: ", e);
        }
        long time1 = System.currentTimeMillis();
        try {
            realTimeDataService.updateUserData(locationInfo);
            realTimeDataService.updateBuildData(locationInfo);
            realTimeDataService.updateUserPoiData(locationInfo);
        } catch (Exception e) {
            logger.error("redis saveInfo error: ", e);
        }
        long time2 = System.currentTimeMillis();
        if (logTag) {
            log("------- hbase=" + (time1 - curTime) + ",redis=" + (time2 - time1));
        }
    }

    private void log(String s) {
        System.out.println(DateUtil.formatCurrentDate("yyyy-MM-dd HH:mm:ss:SSS") + "   " + s);
    }

    private void shutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                log("server stopping... blockingQueue size1:" + blockingQueue.size());
                long start = TimeUtil.currentTimeMillis();
                isRunning = false;
                log("server stopping... callback thread poll ActiveCount1:" + executorService.getActiveCount());
                while (executorService.getActiveCount() > 0 && (TimeUtil.currentTimeMillis() - start < MAX_STOP_TIME)) {
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log("server stopped... blockingQueue size2:" + blockingQueue.size());
                log("server stopped... callback thread poll ActiveCount2:" + executorService.getActiveCount());
            }
        });
    }
}
